package cgl.narada.samples.jms;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.LinkedList;
import java.util.Properties;
import java.util.StringTokenizer;

import javax.jms.BytesMessage;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicPublisher;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;

import cgl.narada.jms.NBJmsInitializer;
import cgl.narada.service.timer.ClockI;
import cgl.narada.util.PropertiesHelp;
import cgl.narada.util.StatUtils;

public class AnabasTest implements javax.jms.MessageListener{
   private TopicSession pubSession;

   private TopicSession subSession;

   private TopicPublisher publisher;

   private TopicConnection connection;

   private String userName;
   
   private int numOfMessagesInBenchmark = 180000;
   private LinkedList lostMessages;
   private LinkedList outOfOrderMessages;
   private LinkedList duplicateMessages;
   private long[] latencies;
   private double[] jitters;
   private int messageNumberReceivedSoFar = 0;
   private int arrayIndex =0;
   private double jitter = 0;
   
   private int MESSAGE_SIZE=1000;
   private long SLEEP_INTERVAL=20;
   
   /** A buffered writer for gathering testing statistics */
   private BufferedWriter buffWrite;
   
   private ClockI hrClock;
   
   private String moduleName = "AnabasTest: ";
   
   private boolean isPublisher = false;
   private long benchmarkStart = 0;
   private long timeOfLastReceipt = 0;
   private boolean benchmarkStarted = false;
   
   public AnabasTest(Properties connectionProperties, String transportType,
               String userName, String topicString) throws Exception {
      NBJmsInitializer ini;
      ini = new NBJmsInitializer(connectionProperties, transportType,
                                 userName.hashCode());
      // Lookup a JMS connection factory
      TopicConnectionFactory conFactory = (TopicConnectionFactory) ini.lookup();
      initializeChatSession(conFactory, userName, topicString);
      
      latencies = new long[numOfMessagesInBenchmark];
      jitters = new double[numOfMessagesInBenchmark];
      
      lostMessages = new LinkedList();
      outOfOrderMessages = new LinkedList();
      duplicateMessages = new LinkedList();
   }
   
   public void initializeChatSession(TopicConnectionFactory conFactory,
                                     String username, String topicString) throws Exception {
      // Create a JMS connection
      TopicConnection connection = conFactory.createTopicConnection("guest",
            "password");

      // Create a JMS session object
      TopicSession pubSession = connection.createTopicSession(false,
            Session.AUTO_ACKNOWLEDGE);
      TopicSession subSession = connection.createTopicSession(false,
            Session.AUTO_ACKNOWLEDGE);

      // Lookup a JMS topic
      Topic chatTopic = pubSession.createTopic(topicString);

      System.out.println("Will be publishing to topic [" + topicString + "]");
      // Create a JMS publisher and subscriber
      TopicPublisher publisher = pubSession.createPublisher(chatTopic);
      TopicSubscriber subscriber = subSession.createSubscriber(chatTopic);

      // Set a JMS message listener
      subscriber.setMessageListener(this);

      // Intialize the Chat application
      set(connection, pubSession, subSession, publisher, username);

      // Start the JMS connection; allows messages to be delivered
      connection.start();
   }

   public void startConnection() throws JMSException {
      connection.start();
   }

   public void stopConnection() throws JMSException {
      connection.stop();
   }

   /* Initializes the instance variables */
   public void set(TopicConnection con, TopicSession pubSess,
                   TopicSession subSess, TopicPublisher pub, String username) {
      this.connection = con;
      this.pubSession = pubSess;
      this.subSession = subSess;
      this.publisher = pub;
      this.userName = username;
   }

   /* Receive message from topic subscriber */
   public void onMessage(Message message) {
      try {
         if (message.getJMSType().equals("BytesMessage")) {
            if (!isPublisher) {
               processBenchmarkMessage(message);
            }
            return;
         }
         TextMessage textMessage = (TextMessage) message;
         String text = textMessage.getText();
         long transitDelay = System.currentTimeMillis() - textMessage.getJMSTimestamp();
         System.out.println(text + ", Delay =" + transitDelay + " mSecs");
      } catch (JMSException jmse) {
         jmse.printStackTrace();
      }
   }

   /* Create and send message using topic publisher */
   public void writeMessage(String text) throws JMSException {
      TextMessage message = pubSession.createTextMessage();
      message.setText("[" + userName + "] : " + text);
      publisher.publish(message);
   }
   
   public Message createMessage(int messageNumber) throws JMSException {
      BytesMessage message = pubSession.createBytesMessage();
      byte[] payload = new byte[MESSAGE_SIZE];
      message.writeBytes(payload);
      message.setIntProperty("Message Number", messageNumber);
 
      return message;
   }
   
   public void processBenchmarkMessage(Message message) throws JMSException {
      if (!benchmarkStarted) {
         benchmarkStarted = true;
         benchmarkStart = System.currentTimeMillis();
      }
      /** Check for message loss and out-of-order delivery*/
      int messageNumber = message.getIntProperty("Message Number");      
      boolean discardMessage = checkAndTrackLostMessages(messageNumber);
      
      if (discardMessage) {
         return;
      }
      
      /** Compute latency **/
      
      latencies[arrayIndex] = System.currentTimeMillis() - message.getJMSTimestamp();
      timeOfLastReceipt =  System.currentTimeMillis();
      /** Compute Jitter **/
      if (arrayIndex >= 1) { 
         jitters[arrayIndex] = 
            StatUtils.computeJitter(jitter, (double)latencies[arrayIndex-1],
                                    (double)latencies[arrayIndex]);
      }
      
      arrayIndex++;
   }
   
   private boolean checkAndTrackLostMessages(int messageNumber) {
      boolean messagesLost = false;
      boolean discardMessage = false;
      
      if (messageNumberReceivedSoFar == messageNumber) {
        duplicateMessages.add(new Integer(messageNumber));
        discardMessage = true;
        return discardMessage;
      }
      
      if (messageNumberReceivedSoFar > messageNumber) {
         Integer key = new Integer(messageNumber);
         if (lostMessages.contains(key)) {
            lostMessages.remove(key);
            outOfOrderMessages.add(key);
         }
         return discardMessage;
      }
      
      if (messageNumberReceivedSoFar + 1 != messageNumber) {
         messagesLost = true;
      }
      
      if (messagesLost) {
         int numOfMessagesLost = messageNumber - (messageNumberReceivedSoFar + 1);
         for (int i=messageNumberReceivedSoFar+1; i < numOfMessagesLost; i++) {
            lostMessages.add(new Integer(i));
         }
      }
      messageNumberReceivedSoFar = messageNumber;
      return discardMessage;
   }
   
   
   public void reportOnBenchmark(String reportFileName, boolean generateReportFile) {
     
      double meanLatency = StatUtils.computeMean(latencies, arrayIndex, 0);
      double standardDeviation = StatUtils.computeStandardDeviation(latencies, arrayIndex, 0);
      
      int numOfLostMessages = lostMessages.size();
      int numOfOutOfOrderMessages = outOfOrderMessages.size();
      
      long timeElapsed = timeOfLastReceipt - benchmarkStart;
      double receivingRate = ((double)arrayIndex/(double)timeElapsed)*1000;
      
      String report = "#Total Number of messages processed within the benchmark = " + arrayIndex + 
        "\n\t#Mean Latency = " + meanLatency + 
        "\n\t#Standard Deviation = " + standardDeviation +
        "\n\t#Number of lost messages = " +  numOfLostMessages +
        "\n\t#Number of Out of Order messages = " + numOfOutOfOrderMessages +
        "\n\t#Receiving rate = " + receivingRate + "messages/sec";
      
      System.out.println(report);
      System.out.println("\t#Pending messages = " + (numOfMessagesInBenchmark - arrayIndex ));
      
      
   
      String lossInfo = "#Lost Message Info: ";
      if (numOfLostMessages != 0 ) {
         for (int i=0; i<numOfLostMessages; i++) {
            lossInfo += ("[" + lostMessages.get(i)+ "] ");
         }
         lossInfo += ("\n#\n#");
         System.out.println(lossInfo);
      }
      
      String outOfOrderInfo = "#Out of Order message Info: ";
      if (numOfOutOfOrderMessages != 0) {
         
         for (int i=0; i<numOfOutOfOrderMessages; i++) {
            outOfOrderInfo += ("[" + outOfOrderMessages.get(i)+ "] ");
         }
      }
      
      
      if (generateReportFile) {
         initializeReportFile(reportFileName);
      
         for (int i=0; i<arrayIndex; i++) {
            String reportLine = i + " " + latencies[i] + " " + jitters[i];
            writeToReportFile(reportLine);
         }
    
         writeToReportFile(report);
         
         if (numOfLostMessages !=0 ) {
            writeToReportFile(lossInfo); 
         }
         
         if (numOfOutOfOrderMessages != 0) {
            writeToReportFile(outOfOrderInfo);
         }
         System.out.println("Report generation complete!");
      }
      
   }
   
   
   
   public void startBenchmarkProcess() throws JMSException {
      isPublisher = true;
      long startTime = System.currentTimeMillis();
      for (int i=1; i <= numOfMessagesInBenchmark; i++) {
         Message message = createMessage(i);
         publisher.publish(message);
         try {
            Thread.sleep(SLEEP_INTERVAL);
         } catch (InterruptedException ie) {
            System.out.println(moduleName + "Error sleeping!");
         }
      }
      long timeElapsed = System.currentTimeMillis() - startTime;
      double publishRate = ((double)numOfMessagesInBenchmark/(double)timeElapsed)*1000;
      
      System.out.println(moduleName + "Benchmark process complete for publisher" +
            "\nNum Of Messages Published = " + numOfMessagesInBenchmark +
            "\nRate of publishing = " + publishRate + " messages/sec");
      
   }

   
   private void initializeReportFile(String fileName) {
      try {
        buffWrite = new BufferedWriter(new FileWriter(new File(fileName)));
        System.out.println(moduleName + "Initialized Subscriber Report File [" + 
          fileName + "]");
      } catch (IOException ioe) {
        System.out.println(moduleName + "Error writing file ...... ");
      } 
    }
    
    
    private void writeToReportFile(String reportLine) {
      try {
        buffWrite.write(reportLine, 0, reportLine.length());
        buffWrite.newLine();
        buffWrite.flush();
      } catch (IOException ioe) {
        System.out.println(moduleName + "Error writing file ...... ");
      } 
    }
   
   
   /* Close the JMS connection */
   public void close() throws JMSException {
      connection.close();
   }

   /* Run the Chat client */
   public static void main(String[] args) {
      String transport = "niotcp";

      try {
         if (args.length != 5) {
            System.out.println("Usage: java cgl.narada.test.Chat <hostname> "
                               + "<portnum> <userName> [niotcp|tcp|udp] <topicString>");
            System.exit(0);
         }

         System.out.println("Type h for usage options");
         Properties props = new Properties();
         /** These properties pertain to setting up a TCP link */
         props.put("hostname", args[0]);
         props.put("portnum", args[1]);

         transport = args[3];
         /**
          * Depending on the transport you choose, you can vary the transport
          * protocol being used. Here we use "tcp". You could also use "udp",
          * "multicast" or "ssl"
          */

         if (transport.equals("http") || transport.equals("ssl")) {
            // String configFileName = "config/BrokerConfiguration.txt";
            String configFileName = "transportTest.properties";
            PropertiesHelp help = new PropertiesHelp(configFileName, transport);
            help.addProperties(props);
         }

         AnabasTest anabas = new AnabasTest(props, transport, args[2], args[4]);

         // read from command line
         BufferedReader commandLine = 
            new BufferedReader(new InputStreamReader(System.in));

         // loop until the word "exit" is typed
         int tracker = 0;
         while (true) {
            String s = commandLine.readLine();
            if (s == null) {
               anabas.close();
               break;
            }
            if (s.equalsIgnoreCase("h")) {
               System.out.println("Simply type messages in for normal chat");
               System.out
                         .println("stop    - Stop the connection. Messages cached");
               System.out
                         .println("start   - Starts connection. Cache released");
               System.out.println("exit    - Exits the application");
               System.out.println("bm      - Start the benchmark");
               System.out.println("pbm <reportFileName>  - Report on benchmark (USE ONLY after benchmark is complete)");
               continue;
            }
            if (s.equalsIgnoreCase("start")) {
               anabas.startConnection();
               continue;
            }
            if (s.equalsIgnoreCase("stop")) {
               anabas.stopConnection();
               continue;
            }
            if (s.equalsIgnoreCase("exit")) {
               anabas.close(); // close down connection
               break;
            }
            
            if (s.equalsIgnoreCase("bm")) {
               anabas.startConnection();
               anabas.startBenchmarkProcess();
               continue;
            }
            
            if (s.startsWith("pbm")) {
               String fileName = null;
               boolean generateReportFile = false;
               
               StringTokenizer tok = new StringTokenizer(s, " ");
               
               if (tok.countTokens() ==2){
                  tok.nextToken();
                  fileName = tok.nextToken();
                  generateReportFile = true;
               }
               
               anabas.reportOnBenchmark(fileName, generateReportFile);
               continue;
            }
            tracker++;
            anabas.writeMessage(s + tracker);
         }
         /** end while-true */
         System.out.println("Chat: exiting ");
      } catch (Exception e) {
         e.printStackTrace();
      }
      System.out.println("Chat: " + Thread.currentThread().getName());
      System.exit(0);

   }
   /** end-main */
  

}
